
package org.example.server.search.listener.mq.kafka;

import cn.hutool.core.lang.TypeReference;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.json.JSONUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Conflicts;
import co.elastic.clients.elasticsearch._types.InlineScript;
import co.elastic.clients.elasticsearch._types.Script;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.UpdateByQueryRequest;
import co.elastic.clients.json.JsonData;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.frw.canal.EntityComposite;
import org.example.frw.canal.HashValue;
import org.example.frw.canal.binlog.BaseCanalBinlog;
import org.example.frw.canal.binlog.entity.GoodsParamBinlogEntity;
import org.example.frw.canal.config.TopicConcurrentlySetting;
import org.example.frw.canal.listener.mq.kafka.BaseConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class GoodsParamConsumer extends BaseConsumer<GoodsParamBinlogEntity> implements BatchAcknowledgingMessageListener<String, String> {
    private static final String TOPIC = "goods__tb_goods_param";
    private static final String GROUP = "search-server";

    @Resource
    private GoodsParamRetryService goodsParamRetryService;

    public GoodsParamConsumer(TopicConcurrentlySetting topicConcurrentlySetting) {
        super(topicConcurrentlySetting, TOPIC ,GROUP);
    }

    /**
     * 幂等由ES保证
     *
     * @param entityComposite entityComposite
     */
    @Override
    @SneakyThrows
    protected void doProcess(EntityComposite<? extends HashValue> entityComposite) {
        log.info("GoodsParamConsumer processing partition={},entityComposite={}", entityComposite.getPartition(), entityComposite);
        GoodsParamBinlogEntity paramEntity = (GoodsParamBinlogEntity) entityComposite.getEntity();
        String scriptSource = null;
        Map<String, JsonData> params = new HashMap<>();
        switch (entityComposite.getType()) {
            case "UPDATE", "INSERT" -> {
                scriptSource = """
                            Map paramMap = new HashMap();
                            if (ctx._source.params != null) {
                               for (int i = 0; i < ctx._source.params.length; i++) {
                                  paramMap.put(ctx._source.params[i].id, ctx._source.params[i]);
                               }
                            }
                            paramMap.put(params.id,params);
                            ctx._source.params = new ArrayList(paramMap.values());
                        """;
                params = getEntityMap(paramEntity);
            }
            case "DELETE" -> {
                scriptSource = """
                             if(ctx._source.params != null){
                                 Map paramMap = new HashMap();
                                 for (Map param : ctx._source.params) {
                                    if(param.get("id") != params.id){
                                        paramMap.put(param.get("id"), param);
                                    }
                                 }
                                 ctx._source.params = new ArrayList(paramMap.values());
                             }
                        """;
                Map<String, JsonData> map = new HashMap<>();
                map.put("id", JsonData.of(paramEntity.getId()));
                params = map;
            }
        }
        if (CharSequenceUtil.isBlank(scriptSource)) {
            log.warn("not supported operation {}", entityComposite.getType());
            return;
        }
        Map<String, JsonData> finalParams = params;
        String finalScriptSource = scriptSource;
        goodsParamRetryService.doUpdateQuery(paramEntity, finalScriptSource, finalParams);
        log.info("processing done！！！");
    }

    private Map<String, JsonData> getEntityMap(GoodsParamBinlogEntity paramEntity) {
        Map<String, JsonData> map = new HashMap<>();
        map.put("id", JsonData.of(paramEntity.getId()));
        map.put("goodsId", JsonData.of(paramEntity.getGoodsId()));
        map.put("name", JsonData.of(paramEntity.getName()));
        map.put("description", JsonData.of(paramEntity.getDescription()));
        map.put("createTime", JsonData.of(paramEntity.createTimeTimestamp()));
        map.put("updateTime", JsonData.of(paramEntity.updateTimeTimestamp()));
        return map;
    }

    @Override
    protected List<EntityComposite<GoodsParamBinlogEntity>> hashIndex(int concurrentlyLevel, ConsumerRecord<String, String> messageExt) {
        log.info("再次分区，concurrentlyLevel={},key={},partition={},offset={}", concurrentlyLevel, messageExt.key(), messageExt.partition(), messageExt.offset());
        String body = messageExt.value();
        TypeReference<BaseCanalBinlog<GoodsParamBinlogEntity>> typeReference = new TypeReference<>() {
        };
        BaseCanalBinlog<GoodsParamBinlogEntity> bean = JSONUtil.toBean(body, typeReference, true);
        return getEntityComposites(concurrentlyLevel, bean);
    }


    @Override
    @KafkaListener(topics = TOPIC, groupId = GROUP, containerFactory = BATCH_CONTAINER_FACTORY)
    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        super.consumeMessage(records, acknowledgment);
    }

    public interface GoodsParamRetryService {
        void doUpdateQuery(GoodsParamBinlogEntity paramEntity, String finalScriptSource, Map<String, JsonData> finalParams) throws IOException;
    }

    @Service
    public static class GoodsParamRetryServiceImpl implements GoodsParamRetryService {
        private final ElasticsearchClient elasticsearchClient;

        public GoodsParamRetryServiceImpl(ElasticsearchClient elasticsearchClient) {
            this.elasticsearchClient = elasticsearchClient;
        }

        @Override
        @Retryable(retryFor = Exception.class, maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 1.5))
        public void doUpdateQuery(GoodsParamBinlogEntity paramEntity, String finalScriptSource, Map<String, JsonData> finalParams) throws IOException {
            UpdateByQueryRequest request = UpdateByQueryRequest.of(
                    builder -> builder
                            .index("goods")
                            .query(Query.of(q ->
                                    q.term(t ->
                                            t.field("id").value(paramEntity.getGoodsId()))
                            ))
                            .script(Script.of(s -> s.inline(
                                    new InlineScript.Builder().lang("painless").source(finalScriptSource).params(finalParams).build()
                            )))
                            .conflicts(Conflicts.Abort)
            );
            log.info("{}", request);
            elasticsearchClient.updateByQuery(request);
        }
    }
}